package com.allyes.flume.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;

//
// How to deal when some line failed to be converted to rawlog?
// Two options:
// drop:
//   (default) just drop that line
// addHeader:
//   Add a key (default: invalid-log) to the event headers with mapping value
//   'yes'. The event body is the description of the error.
//
// Example config:
//   agent.sources.r1.handleInvalidLog.type = addHeader
//   agent.sources.r1.handleInvalidLog.addHeader.key = invalid-log
//

public class InvalidLogHandlerBuild {
  private static final Logger logger = LoggerFactory.getLogger(InvalidLogHandlerBuild.class);

  public static final String HANDLE_INVALID_LOG = "handleInvalidLog";
  public static final String HANDLE_INVALID_LOG_PREFIX = HANDLE_INVALID_LOG + ".";
  public static final String DROP = "drop";
  public static final String DROP_PREFIX = DROP + ".";
  public static final String ADD_HEADER = "addHeader";
  public static final String ADD_HEADER_PREFIX = ADD_HEADER + ".";
  public static final String TYPE = "type";
  public static final String DEFAULT_TYPE = DROP;

  public static InvalidLogHandler build(Context context) {
    if (null == context) {
      context = new Context();
    }

    InvalidLogHandler handler = null;
    Context subContext = new Context();

    String type = context.getString(TYPE, DEFAULT_TYPE);
    switch (type) {
      case DROP:
        handler = new InvalidLogHandlerDrop();
        subContext.putAll(context.getSubProperties(DROP_PREFIX));
        break;

      case ADD_HEADER:
        handler = new InvalidLogHandlerAddHeader();
        subContext.putAll(context.getSubProperties(ADD_HEADER_PREFIX));
        break;

      default:
        logger.error("Unknown type of invalid-log-handler: " + type);
        return null;
    }

    handler.configure(subContext);
    return handler;
  }
}


class InvalidLogHandlerDrop implements InvalidLogHandler {
  @Override
  public Event handle(long inode, String fileName, int lineNum, String line, String errInfo) {
    return null;
  }

  @Override
  public void configure(Context context) {
    // nothing to do
  }
}


class InvalidLogHandlerAddHeader implements InvalidLogHandler {
  public static final String KEY = "key";
  public static final String DEFAULT_KEY = "invalid-log";
  public static final String FIXED_VALUE = "yes";

  private String key;

  @Override
  public void configure(Context context) {
    key = context.getString(KEY, DEFAULT_KEY);
    Preconditions.checkNotNull(key);
    Preconditions.checkState(!key.isEmpty());
  }

  @Override
  public Event handle(long inode, String fileName, int lineNum, String line, String errInfo) {
    String body =
        "Error = " + errInfo +
        ", file = " + fileName +
        ", inode = " + inode +
        ", lineNO. = " + lineNum +
        ", content = " + line;
    Event event = EventBuilder.withBody(body, Charsets.UTF_8);
    event.getHeaders().put(key, FIXED_VALUE);
    return event;
  }
}
